package org.mcclone.jr.spark.hudi;

import org.apache.hudi.DataSourceWriteOptions;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.mcclone.jr.spark.Person;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by McClone on 2020/5/16.
 */
public class HudiSave {

    public static void main(String[] args) {
        SparkSession spark = SparkSession
                .builder()
                .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .master("local[1]")
                .appName("HudiSave")
                .getOrCreate();
        List<Person> personList = new ArrayList<>();
        personList.add(new Person("1111", "1", "1"));
        personList.add(new Person("2222", "1", "2"));
        Dataset<Row> dataFrame = spark.createDataFrame(personList, Person.class);
        dataFrame.write().format("org.apache.hudi")
                .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY(), "type")
                 // 设置数据更新时间的列名
                .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY(), "type")
                 // 设置主键列名
                .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY(), "id")
                .option("hoodie.table.name", "person")
                .option("hoodie.insert.shuffle.parallelism", "2")
//                .option("hoodie.upsert.shuffle.parallelism", "2")
                .mode(SaveMode.Append)
                .save("/data/temp/person");
    }
}
